package lion.framework.core.task.cluster;

import java.util.ArrayList;
import java.util.Date;
import java.util.List;

import org.apache.commons.lang.StringUtils;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;

import lion.dev.lang.MapJ;
import lion.framework.core.db.DB;
import lion.framework.core.db.DataTable;
import lion.framework.core.db.paging.SQLPagingFactory;
import lion.framework.core.task.ISchedule;

/**
 * 数据库集群实例
 *
 * <pre>
 * time   2013年7月26日
 * mail   hl_0074@sina.com
 * </pre>
 *
 * @author lion
 */
public class DBClusterInstance implements ClusterInstance {

	private Log log = LogFactory.getLog(DBClusterInstance.class);

	private String dataSourceName;
	private String instanceId;

	public DBClusterInstance(String dataSourceName, String instanceId) {

		this.dataSourceName = dataSourceName;
		this.instanceId = instanceId;

		saveInstance(instanceId);
	}

	private void saveInstance(String instanceId) {

		try (DB db = new DB(dataSourceName);) {
			boolean b = db.queryTrue("select 1 from clusterinstance where instanceId=?", instanceId);
			if (!b) {
				db.update("insert into clusterinstance(instanceId,checkTime) values(?,now())", instanceId);
			}
		} catch (Exception e) {
			log.error(e);
		}
	}

	@Override
	public void updateStatus() {

		try (DB db = new DB(dataSourceName);) {

			db.update("update clusterinstance set checkTime=now() where instanceId=?", instanceId);
			db.update("delete from clusterinstance where TIMESTAMPDIFF(SECOND,checkTime,now())>=5");
		} catch (Exception e) {
			log.error("", e);
		}
	}

	/**
	 * 查找孤儿日程
	 */
	@Override
	public List<MapJ> findOrphanSchedules() {

		List<MapJ> clustertargets = new ArrayList<>();
		try (DB db = new DB()) {
			findDeadInstanceTargets(db, clustertargets);
			findNoInstanceTargets(db, clustertargets);
		} catch (Exception e) {
			log.error("", e);
		}
		if (log.isDebugEnabled()) {
			log.debug("find orphan tasks: " + clustertargets.size());
		}
		return clustertargets;
	}

	private void findNoInstanceTargets(DB db, List<MapJ> clustertargets) {

		String sql = SQLPagingFactory.paging(db.getType(), "select * from clustertarget ct where not exists (select 1 from clusterInstance ci where ci.instanceId=ct.instanceId)",
				1, 1);
		DataTable dt = db.query(sql);
		for (MapJ row : dt.getRows()) {

			MapJ s = new MapJ();
			s.put("id", row.getString("targetId"));
			s.put("type", row.getString("type", "default"));
			clustertargets.add(s);
			db.update("update clustertarget set instanceId=? where targetId=?", instanceId, row.getString("targetId"));
		}
	}

	private void findDeadInstanceTargets(DB db, List<MapJ> clustertargets) {

		String sql = SQLPagingFactory.paging(db.getType(), "select instanceId,checkTime from clusterinstance where TIMESTAMPDIFF(SECOND,checkTime,now())>5 order by checkTime desc",
				0, 1);
		DataTable dt = db.query(sql);
		if (dt.getRowCount() == 0) { return; }

		for (MapJ instance : dt.getRows()) {

			String id = instance.getString("instanceId");
			if (id == null) {
				continue;
			}
			dt = db.query("select * from clustertarget where instanceId=?", id);
			for (MapJ row : dt.getRows()) {
				MapJ s = new MapJ();
				s.put("id", row.getString("targetId"));
				s.put("type", row.getString("type", "default"));
				clustertargets.add(s);
			}
			db.update("update clustertarget set instanceId=? where instanceId=?", instanceId, id);
		}
	}

	@Override
	public boolean addSchedule(ISchedule schedule) {

		DB db = new DB(dataSourceName);
		db.setTransaction(DB.TRANSACTION_SERIALIZABLE);
		try {
			MapJ target = db.queryOneRow("select instanceId,targetId from clustertarget where targetId=? ", schedule.getId());
			if (target == null) {
				db.save("insert into clustertarget (instanceId,targetId,type) values(?,?,?)", instanceId, schedule.getId(), schedule.getType());
				return true;
			} else {
				String instanceId = target.getString("instanceId");
				if (this.instanceId.equals(instanceId)) { return true; }
				// 查看任务是否有实例在运行，如果没有，则将实例更改为本实例
				Date date = db.queryOne("select checkTime from clusterinstance where instanceId=?", instanceId);
				Date now = new Date();
				// the old instance is dead
				if (date == null || (now.getTime() - date.getTime()) / 1000 > 5) {
					db.update("update clustertarget set instanceId=? where targetId=?", this.instanceId, schedule.getId());
					return true;
				}
			}
		} catch (Exception e) {
			log.error("", e);
			try {
				db.rollback();
			} catch (Exception e1) {
			}
			return false;
		} finally {
			try {
				db.commit();
				db.close();
			} catch (Exception e) {
			}
		}

		return false;
	}

	public void removeSchedule(ISchedule schedule) {

		try (DB db = new DB()) {
			db.update("delete from clustertarget where targetId=?", schedule.getId());
		}
	}

	@Override
	public String getId() {

		return instanceId;
	}

	@Override
	public boolean isScheduleAssigned(ISchedule schedule) {

		// find target and it's instance is alive
		try (DB db = new DB(dataSourceName)) {
			return db.queryTrue(
					"select 1 from clustertarget ct where ct.targetId=? and ct.type=? and exists (select 1 from clusterinstance ci where  TIMESTAMPDIFF(SECOND,ci.checkTime,now())<5 and ci.instanceId=ct.instanceId)",
					schedule.getId(), schedule.getType());
		}
	}

	@Override
	public boolean getLock(String lockName) {

		try (DB db = new DB(dataSourceName)) {
			// transaction goes one by one
			db.setTransaction(DB.TRANSACTION_SERIALIZABLE);

			// clear all of the empty locks
			db.update("delete from clusterlock where instanceId is null");

			String inId = db.queryOne("select instanceId from clusterlock where name=?", lockName);
			// if no instance got the lock
			if (StringUtils.isBlank(inId)) {
				// insert a new lock
				db.update("insert into clusterlock values(?,?)", lockName, instanceId);
				return true;
			}

			// if there is a lock,but the instance is dead
			if (db.queryTrue("select 1 from clusterinstance where  TIMESTAMPDIFF(SECOND,checkTime,now())>5 and instanceId=?", inId)) {
				db.update("update clusterlock set instanceId=? where name=?", instanceId, lockName);
				return true;
			}
		} catch (Exception e) {
			log.error("get lock error", e);
		}
		return false;
	}

	@Override
	public void releaseLock(String lockName) {

		try (DB db = new DB(dataSourceName)) {
			db.setTransaction(DB.TRANSACTION_SERIALIZABLE);
			db.update("delete from clusterlock where instanceId=? and name=?", instanceId, lockName);
		} catch (Exception e) {
			log.error(e);
		}
	}
}
